package com.jiaying.sync;

import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.WakeupException;

import com.jiaying.sync.KafkaMessageParser.SqlStatement;
import com.jiaying.sync.database.SqlServerManager;
import com.jiaying.sync.kafka.KafkaManager;
import com.jiaying.sync.kafka.OffsetManager;

/**
 * Kafka监听器类
 * 用于监听MongoDB变更事件并通过Kafka接收
 * 使用JDK 25虚拟线程优化性能
 */
public class KafkaListener {
    private static final Logger logger = LoggerFactory.getLogger(KafkaListener.class);
    private static final String OFFSET_FILE = "kafka_offsets.properties";
    
    private KafkaManager kafkaManager; // Kafka管理器
    private OffsetManager offsetManager; // 偏移量管理器
    private ExecutorService executorService;
    private ExecutorService messageExecutor;
    private ExecutorService databaseExecutor; // 专用数据库操作线程池
    private ScheduledExecutorService offsetCommitScheduler; // 定时提交偏移量的调度器
    private ScheduledExecutorService batchFlushScheduler; // 定时刷新批量SQL的调度器
    private ScheduledExecutorService statsScheduler; // 定时打印统计信息的调度器
    private AtomicBoolean running = new AtomicBoolean(false);
    private String offsetFilePath;
    private AtomicBoolean shouldCommitOffsets = new AtomicBoolean(false); // 标志位，指示是否需要提交偏移量
    
    // 统计处理数量
    private AtomicInteger totalCount = new AtomicInteger(0);
    private AtomicInteger lastSecondCount = new AtomicInteger(0); // 用于统计每秒处理的数据量
    
    // 批量处理相关的成员变量
    private final BlockingQueue<SqlStatement> insertBatch = new LinkedBlockingQueue<>(10000); // 增大队列大小
    private final BlockingQueue<SqlStatement> updateBatch = new LinkedBlockingQueue<>(10000); // 增大队列大小
    private static final int BATCH_SIZE = 200; // 增加批量大小到1000
    private static final int BATCH_FLUSH_INTERVAL = 5; // 批量刷新间隔（秒），保持5秒
    
    // 按集合名称分组的批量队列
    private final Map<String, BlockingQueue<SqlStatement>> collectionInsertBatches = new HashMap<>();
    private final Map<String, BlockingQueue<SqlStatement>> collectionUpdateBatches = new HashMap<>();
    
    // SQL Server管理器
    private SqlServerManager sqlServerManager;
    
    /**
     * 构造函数
     */
    public KafkaListener() {
        // 初始化偏移量管理器
        initializeOffsetManager();
        
        // 初始化Kafka管理器
        initializeKafkaManager();
        
        // 初始化SQL Server管理器
        initializeSqlServerManager();
        if (!testSqlConnection()) {
            System.exit(0);
        }
        
        // 初始化偏移量文件路径
        this.offsetFilePath = OFFSET_FILE;
        
        // 使用虚拟线程优化线程池
        this.executorService = Executors.newVirtualThreadPerTaskExecutor();
        this.messageExecutor = Executors.newVirtualThreadPerTaskExecutor();
        this.databaseExecutor = Executors.newVirtualThreadPerTaskExecutor(); // 专用数据库操作线程池
        this.offsetCommitScheduler = Executors.newSingleThreadScheduledExecutor();
        this.batchFlushScheduler = Executors.newSingleThreadScheduledExecutor();
        this.statsScheduler = Executors.newSingleThreadScheduledExecutor();
    }
    
    /**
     * 初始化偏移量管理器
     */
    private void initializeOffsetManager() {
        try {
            this.offsetManager = new OffsetManager(OFFSET_FILE);
            logger.info("偏移量管理器初始化成功");
        } catch (Exception e) {
            logger.error("偏移量管理器初始化失败: {}", e.getMessage(), e);
        }
    }
    
    /**
     * 初始化Kafka管理器
     */
    private void initializeKafkaManager() {
        try {
            this.kafkaManager = new KafkaManager();
            logger.info("Kafka管理器初始化成功");
        } catch (Exception e) {
            logger.error("Kafka管理器初始化失败: {}", e.getMessage(), e);
        }
    }
    
    /**
     * 初始化SQL Server管理器
     */
    private void initializeSqlServerManager() {
        try {
            this.sqlServerManager = new SqlServerManager();
            logger.info("SQL Server管理器初始化成功");
        } catch (Exception e) {
            logger.error("SQL Server管理器初始化失败: {}", e.getMessage(), e);
        }
    }
    
    /**
     * 测试SQL Server连接
     */
    public boolean testSqlConnection() {
        if (sqlServerManager != null) {
            return sqlServerManager.testConnection();
        }
        return false;
    }
    
    /**
     * 从连接池获取连接
     * @return SQL Server连接
     * @throws SQLException
     */
    private Connection getConnection() throws SQLException {
        if (sqlServerManager != null) {
            return sqlServerManager.getConnection();
        } else {
            throw new SQLException("连接池未初始化");
        }
    }
    
    /**
     * 获取指定集合的INSERT批量队列（不存在则创建）
     * @param collectionName 集合名称
     * @return 该集合的INSERT批量队列
     */
    private BlockingQueue<SqlStatement> getCollectionInsertQueue(String collectionName) {
        return collectionInsertBatches.computeIfAbsent(collectionName, k -> new LinkedBlockingQueue<>(1000));
    }
    
    /**
     * 获取指定集合的UPDATE批量队列（不存在则创建）
     * @param collectionName 集合名称
     * @return 该集合的UPDATE批量队列
     */
    private BlockingQueue<SqlStatement> getCollectionUpdateQueue(String collectionName) {
        return collectionUpdateBatches.computeIfAbsent(collectionName, k -> new LinkedBlockingQueue<>(1000));
    }
    
    /**
     * 订阅主题
     * @param topicPattern 主题模式（正则表达式）
     */
    public void subscribe(String topicPattern) {
        if (kafkaManager != null) {
            kafkaManager.subscribe(topicPattern, new ConsumerRebalanceListener() {
                @Override
                public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                    // 分区被撤销时提交偏移量
                    commitOffsets();
                }
                
                @Override
                public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                    // 分区被分配时加载偏移量
                    loadOffsets(partitions);
                }
            });
        }
    }
    
    /**
     * 开始监听
     */
    public void startListening() {
        if (running.compareAndSet(false, true)) {
            executorService.submit(() -> {
                logger.info("开始监听Kafka消息...");
                try {
                    // 启动定时提交偏移量任务，每5秒执行一次
                    offsetCommitScheduler.scheduleAtFixedRate(() -> shouldCommitOffsets.set(true), 5, 5, TimeUnit.SECONDS);
                    
                    // 启动定时刷新批量SQL任务，每5秒执行一次
                    batchFlushScheduler.scheduleAtFixedRate(this::flushBatchStatements, BATCH_FLUSH_INTERVAL, BATCH_FLUSH_INTERVAL, TimeUnit.SECONDS);
                    
                    // 启动定时打印统计信息任务，每秒执行一次
                    statsScheduler.scheduleAtFixedRate(this::printStats, 1, 1, TimeUnit.SECONDS);
                    
                    while (running.get()) {
                        try {
                            ConsumerRecords<String, String> records = kafkaManager.poll(100); // 减少poll超时时间到100ms
                            for (ConsumerRecord<String, String> record : records) {
                                // 提交消息到处理线程池
                                messageExecutor.submit(() -> handleMessage(record));
                            }
                            
                            // 检查是否需要提交偏移量
                            if (shouldCommitOffsets.compareAndSet(true, false)) {
                                commitOffsets();
                            }
                        } catch (WakeupException e) {
                            // 忽略唤醒异常
                            if (running.get()) {
                                throw e;
                            }
                        } catch (Exception e) {
                            logger.error("消费消息时发生错误: {}", e.getMessage(), e);
                        }
                    }
                } finally {
                    // 提交最后的偏移量
                    commitOffsets();
                    // 执行剩余的批量SQL语句
                    flushBatchStatements();
                }
            });
        }
    }
    
    /**
     * 停止监听
     */
    public void stopListening() {
        logger.info("正在停止Kafka监听器...");
        if (running.compareAndSet(true, false)) {
            // 唤醒消费者
            if (kafkaManager != null) {
                kafkaManager.wakeup();
            }
            
            // 执行剩余的批量SQL语句
            flushBatchStatements();
            
            // 关闭执行器
            try {
                // 关闭所有线程池
                messageExecutor.shutdown();
                executorService.shutdown();
                databaseExecutor.shutdown(); // 关闭数据库专用线程池
                offsetCommitScheduler.shutdown();
                batchFlushScheduler.shutdown();
                if (statsScheduler != null) {
                    statsScheduler.shutdown(); // 关闭统计信息调度器
                }
                
                // 等待各线程池完成关闭
                if (!executorService.awaitTermination(5, java.util.concurrent.TimeUnit.SECONDS)) {
                    logger.warn("主监听线程池关闭超时，强制关闭所有线程池");
                    executorService.shutdownNow();
                }
                
                if (!messageExecutor.awaitTermination(10, java.util.concurrent.TimeUnit.SECONDS)) {
                    logger.warn("消息处理器线程池关闭超时，强制关闭");
                    messageExecutor.shutdownNow();
                }
                
                if (!databaseExecutor.awaitTermination(5, java.util.concurrent.TimeUnit.SECONDS)) {
                    logger.warn("数据库操作线程池关闭超时，强制关闭");
                    databaseExecutor.shutdownNow();
                }
                
                if (!offsetCommitScheduler.awaitTermination(5, java.util.concurrent.TimeUnit.SECONDS)) {
                    logger.warn("偏移量提交调度器关闭超时，强制关闭");
                    offsetCommitScheduler.shutdownNow();
                }
                
                if (!batchFlushScheduler.awaitTermination(5, java.util.concurrent.TimeUnit.SECONDS)) {
                    logger.warn("批量刷新调度器关闭超时，强制关闭");
                    batchFlushScheduler.shutdownNow();
                }
                
                if (statsScheduler != null && !statsScheduler.awaitTermination(5, java.util.concurrent.TimeUnit.SECONDS)) {
                    logger.warn("统计信息调度器关闭超时，强制关闭");
                    statsScheduler.shutdownNow();
                }
            } catch (InterruptedException e) {
                logger.error("关闭线程池时被中断，强制关闭所有线程池");
                messageExecutor.shutdownNow();
                executorService.shutdownNow();
                databaseExecutor.shutdownNow(); // 强制关闭数据库专用线程池
                offsetCommitScheduler.shutdownNow();
                batchFlushScheduler.shutdownNow();
                if (statsScheduler != null) {
                    statsScheduler.shutdownNow(); // 强制关闭统计信息调度器
                }
                Thread.currentThread().interrupt();
            }
            
            // 关闭SQL Server连接池
            if (sqlServerManager != null) {
                sqlServerManager.close();
                logger.info("SQL Server连接池已关闭");
            }
        }
        
        logger.info("Kafka监听器已停止");
    }
    
    /**
     * 执行剩余的批量SQL语句（包括按集合分组的队列）
     */
    private void flushBatchStatements() {
        // 执行按集合分组的剩余INSERT语句
        for (Map.Entry<String, BlockingQueue<SqlStatement>> entry : collectionInsertBatches.entrySet()) {
            String collectionName = entry.getKey();
            BlockingQueue<SqlStatement> collectionQueue = entry.getValue();
            
            if (!collectionQueue.isEmpty()) {
                List<SqlStatement> remainingInserts = new ArrayList<>();
                collectionQueue.drainTo(remainingInserts);
                
                if (!remainingInserts.isEmpty()) {
                    databaseExecutor.submit(() -> executeBatchInsertsByCollection(collectionName, remainingInserts));
                }
            }
        }
        
        // 执行按集合分组的剩余UPDATE语句
        for (Map.Entry<String, BlockingQueue<SqlStatement>> entry : collectionUpdateBatches.entrySet()) {
            String collectionName = entry.getKey();
            BlockingQueue<SqlStatement> collectionQueue = entry.getValue();
            
            if (!collectionQueue.isEmpty()) {
                List<SqlStatement> remainingUpdates = new ArrayList<>();
                collectionQueue.drainTo(remainingUpdates);
                
                if (!remainingUpdates.isEmpty()) {
                    databaseExecutor.submit(() -> executeBatchUpdatesByCollection(collectionName, remainingUpdates));
                }
            }
        }
        
        // 执行原始全局队列的剩余语句（向后兼容性）
        // 执行剩余的INSERT语句
        if (!insertBatch.isEmpty()) {
            List<SqlStatement> remainingInserts = new ArrayList<>();
            insertBatch.drainTo(remainingInserts);
            
            if (!remainingInserts.isEmpty()) {
                executeBatchInserts(remainingInserts);
            }
        }
        
        // 执行剩余的UPDATE语句
        if (!updateBatch.isEmpty()) {
            List<SqlStatement> remainingUpdates = new ArrayList<>();
            updateBatch.drainTo(remainingUpdates);
            
            if (!remainingUpdates.isEmpty()) {
                executeBatchUpdates(remainingUpdates);
            }
        }
    }
    
    /**
     * 处理接收到的消息
     * @param record Kafka消息记录
     */
    protected void handleMessage(ConsumerRecord<String, String> record) {
        // System.out.printf("msg - topic: %s, partition: %d, offset: %d, key: %s, value: %s%n",
        //         record.topic(), record.partition(), record.offset(), record.key(), record.value());
        // 增加处理计数
        totalCount.incrementAndGet();
        // 解析消息并提取关键信息，同时传入key参数
        try {
            // KafkaMessageParser.ParsedMessage parsedMessage=null;
            KafkaMessageParser.ParsedMessage parsedMessage = KafkaMessageParser.parseMessage(record.key(), record.value());
            if (parsedMessage != null) {
                logger.debug("解析结果:");
                logger.debug("  操作符: {}", parsedMessage.getOperation());
                logger.debug("  表名: {}", parsedMessage.getCollectionName());
                logger.debug("  主键ID: {}", parsedMessage.getId());
                
                // 获取集合名称
                String collectionName = parsedMessage.getCollectionName();
                
                // 获取生成的SQL语句
                SqlStatement insertSQL = parsedMessage.getInsertSqlStatement();
                SqlStatement updateSQL = parsedMessage.getUpdateSqlStatement();
                
                // 按集合名称分组执行SQL语句（优化后的方法）
                executeSqlStatementsByCollection(collectionName, insertSQL, updateSQL);
                
                // 定期输出队列状态，用于监控（包括按集合分组的队列）
                int totalInsertSize = collectionInsertBatches.values().stream().mapToInt(BlockingQueue::size).sum();
                int totalUpdateSize = collectionUpdateBatches.values().stream().mapToInt(BlockingQueue::size).sum();
                
                if (totalInsertSize > 500 || totalUpdateSize > 500) {
                    logger.warn("批量队列积压严重 - INSERT队列: {}, UPDATE队列: {}", totalInsertSize, totalUpdateSize);
                    logger.warn("各集合队列详情:");
                    collectionInsertBatches.forEach((collection, queue) -> {
                        if (queue.size() > 0) {
                            logger.warn("  {} - INSERT: {}", collection, queue.size());
                        }
                    });
                    collectionUpdateBatches.forEach((collection, queue) -> {
                        if (queue.size() > 0) {
                            logger.warn("  {} - UPDATE: {}", collection, queue.size());
                        }
                    });
                }
            }
        } catch (Exception e) {
            logger.error("解析Kafka消息时发生错误: {}", e.getMessage(), e);
        }
    }
    
    /**
     * 按集合名称分组执行SQL语句（性能优化版本）
     * @param collectionName 集合名称
     * @param insertSQL INSERT语句
     * @param updateSQL UPDATE语句
     */
    private void executeSqlStatementsByCollection(String collectionName, SqlStatement insertSQL, SqlStatement updateSQL) {
        // 将SQL语句添加到对应集合的批量处理队列中
        if (insertSQL != null) {
            try {
                BlockingQueue<SqlStatement> collectionInsertQueue = getCollectionInsertQueue(collectionName);
                // 使用阻塞方式添加，当队列满时会阻塞，提供背压控制
                collectionInsertQueue.put(insertSQL);
                
                // 检查是否达到批量大小
                if (collectionInsertQueue.size() >= BATCH_SIZE) {
                    // 获取一批数据进行处理
                    List<SqlStatement> batch = new ArrayList<>(BATCH_SIZE);
                    collectionInsertQueue.drainTo(batch, BATCH_SIZE);
                    
                    if (!batch.isEmpty()) {
                        // 使用专用数据库线程池异步执行批量INSERT
                        String finalCollectionName = collectionName;
                        databaseExecutor.submit(() -> executeBatchInsertsByCollection(finalCollectionName, batch));
                    }
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                logger.error("添加INSERT语句到集合 {} 队列时被中断", collectionName, e);
            }
        }
        
        if (updateSQL != null) {
            try {
                BlockingQueue<SqlStatement> collectionUpdateQueue = getCollectionUpdateQueue(collectionName);
                // 使用阻塞方式添加，当队列满时会阻塞，提供背压控制
                collectionUpdateQueue.put(updateSQL);
                
                // 检查是否达到批量大小
                if (collectionUpdateQueue.size() >= BATCH_SIZE) {
                    // 获取一批数据进行处理
                    List<SqlStatement> batch = new ArrayList<>(BATCH_SIZE);
                    collectionUpdateQueue.drainTo(batch, BATCH_SIZE);
                    
                    if (!batch.isEmpty()) {
                        // 使用专用数据库线程池异步执行批量UPDATE
                        String finalCollectionName = collectionName;
                        databaseExecutor.submit(() -> executeBatchUpdatesByCollection(finalCollectionName, batch));
                    }
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                logger.error("添加UPDATE语句到集合 {} 队列时被中断", collectionName, e);
            }
        }
    }
    
    /**
     * 执行SQL语句（原始方法，保留用于兼容性）
     * @param insertSQL INSERT语句
     * @param updateSQL UPDATE语句
     */
    private void executeSqlStatements(SqlStatement insertSQL, SqlStatement updateSQL) {
        // 将SQL语句添加到批量处理队列中
        if (insertSQL != null) {
            try {
                // 使用阻塞方式添加，当队列满时会阻塞，提供背压控制
                insertBatch.put(insertSQL);
                
                // 检查是否达到批量大小
                if (insertBatch.size() >= BATCH_SIZE) {
                    // 获取一批数据进行处理
                    List<SqlStatement> batch = new ArrayList<>(BATCH_SIZE);
                    insertBatch.drainTo(batch, BATCH_SIZE);
                    
                    if (!batch.isEmpty()) {
                        // 使用专用数据库线程池异步执行批量INSERT
                        databaseExecutor.submit(() -> executeBatchInserts(batch));
                    }
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                logger.error("添加INSERT语句到队列时被中断", e);
            }
        }
        
        if (updateSQL != null) {
            try {
                // 使用阻塞方式添加，当队列满时会阻塞，提供背压控制
                updateBatch.put(updateSQL);
                
                // 检查是否达到批量大小
                if (updateBatch.size() >= BATCH_SIZE) {
                    // 获取一批数据进行处理
                    List<SqlStatement> batch = new ArrayList<>(BATCH_SIZE);
                    updateBatch.drainTo(batch, BATCH_SIZE);
                    
                    if (!batch.isEmpty()) {
                        // 使用专用数据库线程池异步执行批量UPDATE
                        databaseExecutor.submit(() -> executeBatchUpdates(batch));
                    }
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                logger.error("添加UPDATE语句到队列时被中断", e);
            }
        }
    }
    
    /**
     * 按集合执行批量INSERT语句（性能优化版本）
     * @param collectionName 集合名称
     * @param insertStatements INSERT语句列表
     */
    private void executeBatchInsertsByCollection(String collectionName, List<SqlStatement> insertStatements) {
        if (insertStatements.isEmpty()) {
            return;
        }
        
        // 检查连接池是否已关闭
        if (sqlServerManager == null || sqlServerManager.isClosed()) {
            logger.error("SQL Server连接池已关闭，无法执行集合 {} 的批量INSERT语句", collectionName);
            return;
        }
        
        // 按SQL语句模板分组（在同一集合内）
        Map<String, List<SqlStatement>> groupedStatements = new HashMap<>();
        for (SqlStatement statement : insertStatements) {
            groupedStatements.computeIfAbsent(statement.getSql(), k -> new ArrayList<>()).add(statement);
        }
        
        Connection conn = null;
        try {
            // 从连接池获取连接，同一批次复用连接
            conn = getConnection();
            conn.setAutoCommit(false); // 手动控制事务
            
            int totalProcessed = 0;
            
            // 对每组相同的SQL语句进行批量执行
            for (Map.Entry<String, List<SqlStatement>> entry : groupedStatements.entrySet()) {
                String sqlTemplate = entry.getKey();
                List<SqlStatement> statements = entry.getValue();
                
                PreparedStatement pstmt = null;
                try {
                    // 创建PreparedStatement
                    pstmt = conn.prepareStatement(sqlTemplate);
                    
                    // 添加所有参数到批处理中
                    for (SqlStatement sqlStmt : statements) {
                        setParameters(pstmt, sqlStmt.getParameters());
                        pstmt.addBatch();
                    }
                    
                    // 执行批处理
                    int[] results = pstmt.executeBatch();
                    totalProcessed += results.length;
                    
                    // 清理资源
                    if (pstmt != null) {
                        pstmt.close();
                    }
                } catch (SQLException e) {
                    logger.error("执行集合 {} 批量INSERT语句失败: {}", collectionName, e.getMessage(), e);
                    logger.error("SQL: {}", sqlTemplate);
                    logger.error("参数: {}", statements.toString());
                    // 发生错误时回滚事务
                    conn.rollback();
                    return;
                }
            }
            
            // 提交事务
            conn.commit();
            logger.info("集合 {} 批量INSERT执行成功，共插入 {} 条记录", collectionName, totalProcessed);
        } catch (SQLException e) {
            logger.error("执行集合 {} 批量INSERT事务失败: {}", collectionName, e.getMessage(), e);
        } finally {
            // 关闭资源
            if (conn != null) {
                try {
                    conn.setAutoCommit(true); // 恢复自动提交
                    conn.close(); // 返回连接到池中
                } catch (SQLException e) {
                    logger.error("关闭Connection失败: {}", e.getMessage(), e);
                }
            }
        }
    }
    
    /**
     * 执行批量INSERT语句（原始方法，保留用于兼容性）
     * @param insertStatements INSERT语句列表
     */
    private void executeBatchInserts(List<SqlStatement> insertStatements) {
        if (insertStatements.isEmpty()) {
            return;
        }
        
        // 检查连接池是否已关闭
        if (sqlServerManager == null || sqlServerManager.isClosed()) {
            logger.error("SQL Server连接池已关闭，无法执行批量INSERT语句");
            return;
        }
        
        // 按SQL语句模板分组
        Map<String, List<SqlStatement>> groupedStatements = new HashMap<>();
        for (SqlStatement statement : insertStatements) {
            groupedStatements.computeIfAbsent(statement.getSql(), k -> new ArrayList<>()).add(statement);
        }
        
        Connection conn = null;
        try {
            // 从连接池获取连接，同一批次复用连接
            conn = getConnection();
            conn.setAutoCommit(false); // 手动控制事务
            
            // 对每组相同的SQL语句进行批量执行
            for (Map.Entry<String, List<SqlStatement>> entry : groupedStatements.entrySet()) {
                String sqlTemplate = entry.getKey();
                List<SqlStatement> statements = entry.getValue();
                
                PreparedStatement pstmt = null;
                try {
                    // 创建PreparedStatement
                    pstmt = conn.prepareStatement(sqlTemplate);
                    
                    // 添加所有参数到批处理中
                    for (SqlStatement sqlStmt : statements) {
                        setParameters(pstmt, sqlStmt.getParameters());
                        pstmt.addBatch();
                    }
                    
                    // 执行批处理
                    int[] results = pstmt.executeBatch();
                    logger.info("批量INSERT执行成功，共插入 {} 条记录", results.length);
                    
                    // 清理资源
                    if (pstmt != null) {
                        pstmt.close();
                    }
                } catch (SQLException e) {
                    logger.error("执行批量INSERT语句失败: {}", e.getMessage(), e);
                    logger.error("SQL: {}", sqlTemplate);
                    logger.error("参数: {}", statements.toString());
                    // 发生错误时回滚事务
                    conn.rollback();
                    return;
                }
            }
            
            // 提交事务
            conn.commit();
        } catch (SQLException e) {
            logger.error("执行批量INSERT事务失败: {}", e.getMessage(), e);
        } finally {
            // 关闭资源
            if (conn != null) {
                try {
                    conn.setAutoCommit(true); // 恢复自动提交
                    conn.close(); // 返回连接到池中
                } catch (SQLException e) {
                    logger.error("关闭Connection失败: {}", e.getMessage(), e);
                }
            }
        }
    }
    
    /**
     * 按集合执行批量UPDATE语句（性能优化版本）
     * @param collectionName 集合名称
     * @param updateStatements UPDATE语句列表
     */
    private void executeBatchUpdatesByCollection(String collectionName, List<SqlStatement> updateStatements) {
        if (updateStatements.isEmpty()) {
            return;
        }
        
        // 检查连接池是否已关闭
        if (sqlServerManager == null || sqlServerManager.isClosed()) {
            logger.error("SQL Server连接池已关闭，无法执行集合 {} 的批量UPDATE语句", collectionName);
            return;
        }
        
        // 按SQL语句模板分组（在同一集合内）
        Map<String, List<SqlStatement>> groupedStatements = new HashMap<>();
        for (SqlStatement statement : updateStatements) {
            groupedStatements.computeIfAbsent(statement.getSql(), k -> new ArrayList<>()).add(statement);
        }
        
        Connection conn = null;
        try {
            // 从连接池获取连接，同一批次复用连接
            conn = getConnection();
            conn.setAutoCommit(false); // 手动控制事务
            
            int totalProcessed = 0;
            
            // 对每组相同的SQL语句进行批量执行
            for (Map.Entry<String, List<SqlStatement>> entry : groupedStatements.entrySet()) {
                String sqlTemplate = entry.getKey();
                List<SqlStatement> statements = entry.getValue();
                
                PreparedStatement pstmt = null;
                try {
                    // 创建PreparedStatement
                    pstmt = conn.prepareStatement(sqlTemplate);
                    
                    // 添加所有参数到批处理中
                    for (SqlStatement sqlStmt : statements) {
                        setParameters(pstmt, sqlStmt.getParameters());
                        pstmt.addBatch();
                    }
                    
                    // 执行批处理
                    int[] results = pstmt.executeBatch();
                    totalProcessed += results.length;
                    
                    // 清理资源
                    if (pstmt != null) {
                        pstmt.close();
                    }
                } catch (SQLException e) {
                    logger.error("执行集合 {} 批量UPDATE语句失败: {}", collectionName, e.getMessage(), e);
                    logger.error("SQL: {}", sqlTemplate);
                    logger.error("参数: {}", statements.toString());
                    // 发生错误时回滚事务
                    conn.rollback();
                    return;
                }
            }
            
            // 提交事务
            conn.commit();
            logger.info("集合 {} 批量UPDATE执行成功，共更新 {} 条记录", collectionName, totalProcessed);
        } catch (SQLException e) {
            logger.error("执行集合 {} 批量UPDATE事务失败: {}", collectionName, e.getMessage(), e);
        } finally {
            // 关闭资源
            if (conn != null) {
                try {
                    conn.setAutoCommit(true); // 恢复自动提交
                    conn.close(); // 返回连接到池中
                } catch (SQLException e) {
                    logger.error("关闭Connection失败: {}", e.getMessage(), e);
                }
            }
        }
    }
    
    /**
     * 执行批量UPDATE语句（原始方法，保留用于兼容性）
     * @param updateStatements UPDATE语句列表
     */
    private void executeBatchUpdates(List<SqlStatement> updateStatements) {
        if (updateStatements.isEmpty()) {
            return;
        }
        
        // 检查连接池是否已关闭
        if (sqlServerManager == null || sqlServerManager.isClosed()) {
            logger.error("SQL Server连接池已关闭，无法执行批量UPDATE语句");
            return;
        }
        
        // 按SQL语句模板分组
        Map<String, List<SqlStatement>> groupedStatements = new HashMap<>();
        for (SqlStatement statement : updateStatements) {
            groupedStatements.computeIfAbsent(statement.getSql(), k -> new ArrayList<>()).add(statement);
        }
        
        Connection conn = null;
        try {
            // 从连接池获取连接，同一批次复用连接
            conn = getConnection();
            conn.setAutoCommit(false); // 手动控制事务
            
            // 对每组相同的SQL语句进行批量执行
            for (Map.Entry<String, List<SqlStatement>> entry : groupedStatements.entrySet()) {
                String sqlTemplate = entry.getKey();
                List<SqlStatement> statements = entry.getValue();
                
                PreparedStatement pstmt = null;
                try {
                    // 创建PreparedStatement
                    pstmt = conn.prepareStatement(sqlTemplate);
                    
                    // 添加所有参数到批处理中
                    for (SqlStatement sqlStmt : statements) {
                        setParameters(pstmt, sqlStmt.getParameters());
                        pstmt.addBatch();
                    }
                    
                    // 执行批处理
                    int[] results = pstmt.executeBatch();
                    logger.info("批量UPDATE执行成功，共更新 {} 条记录", results.length);
                    
                    // 清理资源
                    if (pstmt != null) {
                        pstmt.close();
                    }
                } catch (SQLException e) {
                    logger.error("执行批量UPDATE语句失败: {}", e.getMessage(), e);
                    logger.error("SQL: {}", sqlTemplate);
                    // 发生错误时回滚事务
                    conn.rollback();
                    return;
                }
            }
            
            // 提交事务
            conn.commit();
        } catch (SQLException e) {
            logger.error("执行批量UPDATE事务失败: {}", e.getMessage(), e);
        } finally {
            // 关闭资源
            if (conn != null) {
                try {
                    conn.setAutoCommit(true); // 恢复自动提交
                    conn.close(); // 返回连接到池中
                } catch (SQLException e) {
                    logger.error("关闭Connection失败: {}", e.getMessage(), e);
                }
            }
        }
    }
    
    /**
     * 设置PreparedStatement参数
     * @param statement PreparedStatement
     * @param parameters 参数数组
     * @throws SQLException
     */
    private void setParameters(PreparedStatement statement, Object[] parameters) throws SQLException {
        if (parameters != null) {
            for (int i = 0; i < parameters.length; i++) {
                Object param = parameters[i];
                int paramIndex = i + 1;
                
                // 处理null值
                if (param == null) {
                    statement.setNull(paramIndex, java.sql.Types.VARCHAR);
                    continue;
                }
                
                // 根据参数类型设置值，避免类型不明确的问题
                if (param instanceof String) {
                    statement.setString(paramIndex, (String) param);
                } else if (param instanceof Integer) {
                    statement.setInt(paramIndex, (Integer) param);
                } else if (param instanceof Long) {
                    statement.setLong(paramIndex, (Long) param);
                } else if (param instanceof Double) {
                    statement.setDouble(paramIndex, (Double) param);
                } else if (param instanceof Float) {
                    statement.setFloat(paramIndex, (Float) param);
                } else if (param instanceof Boolean) {
                    statement.setBoolean(paramIndex, (Boolean) param);
                } else if (param instanceof Timestamp) {
                    statement.setTimestamp(paramIndex, (Timestamp) param);
                } else if (param instanceof java.sql.Date) {
                    statement.setDate(paramIndex, (java.sql.Date) param);
                } else if (param instanceof java.util.Date) {
                    statement.setTimestamp(paramIndex, new Timestamp(((java.util.Date) param).getTime()));
                } else if (param instanceof BigDecimal) {
                    statement.setBigDecimal(paramIndex, (BigDecimal) param);
                } else {
                    // 对于其他类型，使用setString以避免类型转换问题
                    statement.setString(paramIndex, param.toString());
                }
            }
        }
    }
    
    /**
     * 提交偏移量到文件
     */
    private void commitOffsets() {
        try {
            Map<TopicPartition, Long> offsets = new HashMap<TopicPartition, Long>();
            if (kafkaManager != null) {
                for (TopicPartition partition : kafkaManager.assignment()) {
                    offsets.put(partition, kafkaManager.position(partition));
                }
            }
            
            Map<String, Long> offsetMap = new HashMap<String, Long>();
            for (Map.Entry<TopicPartition, Long> entry : offsets.entrySet()) {
                TopicPartition partition = entry.getKey();
                long offset = entry.getValue();
                String key = partition.topic() + "-" + partition.partition();
                offsetMap.put(key, offset);
            }
            
            // 只有当偏移量发生变化时才保存到文件
            if (offsetManager != null && offsetManager.hasOffsetChanged(offsetMap)) {
                // 保存到文件
                offsetManager.saveOffsetsToFile(offsetMap);
                logger.info("偏移量已保存到文件: {}", offsetFilePath);
            }
        } catch (Exception e) {
            logger.error("保存偏移量时出错: {}", e.getMessage(), e);
        }
    }
    
    /**
     * 检查偏移量是否发生变化
     * @param currentOffsets 当前偏移量
     * @return 是否发生变化
     */
    private boolean hasOffsetChanged(Map<String, Long> currentOffsets) {
        if (offsetManager != null) {
            return offsetManager.hasOffsetChanged(currentOffsets);
        }
        return true;
    }
    
    /**
     * 从文件加载偏移量
     * @param partitions 分区列表
     */
    private void loadOffsets(Collection<TopicPartition> partitions) {
        if (offsetManager != null && kafkaManager != null) {
            offsetManager.loadOffsets(partitions, kafkaManager::seek);
        }
    }
    
    /**
     * 保存偏移量到文件
     * @param offsetMap 偏移量映射
     */
    private void saveOffsetsToFile(Map<String, Long> offsetMap) {
        // 此方法已移至OffsetManager类中
    }
    
    /**
     * 从文件加载偏移量
     * @return 偏移量映射
     */
    private Map<String, Long> loadOffsetsFromFile() {
        // 此方法已移至OffsetManager类中
        return new HashMap<>();
    }
    
    /**
     * 连接字符串列表
     * @param list 字符串列表
     * @param separator 分隔符
     * @return 连接后的字符串
     */
    private String join(List<String> list, String separator) {
        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < list.size(); i++) {
            sb.append(list.get(i));
            if (i < list.size() - 1) {
                sb.append(separator);
            }
        }
        return sb.toString();
    }
    
    /**
     * 打印统计信息
     */
    private void printStats() {
        int currentCount = totalCount.get();
        int lastCount = lastSecondCount.getAndSet(currentCount);
        int perSecond = currentCount - lastCount;
        logger.info("总计处理: {} 条, 本秒处理: {} 条", currentCount, perSecond);
    }
    
    /**
     * 主方法 - 演示Kafka监听器的使用
     */
    public static void main(String[] args) throws InterruptedException {
        // 创建Kafka监听器实例
        KafkaListener listener = new KafkaListener();
        
        // 添加关闭钩子以实现优雅关闭
        Runtime.getRuntime().addShutdownHook(new Thread() {
            @Override
            public void run() {
                listener.stopListening();
            }
        });
        
        try {
            // 订阅主题模式（使用正则表达式）
            // 示例：监听mongo-testdb开头的所有主题
            listener.subscribe("mongo-testdb\\..*");
            
            // 开始监听
            listener.startListening();
            
            logger.info("Kafka监听器正在运行，按Ctrl+C停止...");
            
            // 保持主线程运行
            Thread.currentThread().join();
        } catch (InterruptedException e) {
            logger.info("主线程被中断");
        } finally {
            listener.stopListening();
        }
    }
}